package org.redis.cluster.registry;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.redis.cluster.Configs;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.config.RegistryConfig;
import org.redis.cluster.config.ServerConfig;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.registry.exception.RegistryException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;

public abstract class AbstractRegistry implements Registry,Plugin {

	private final static NodeMode DEFAULT_NODE_MODE=NodeMode.PERSISTENT;
	
	private final static NodeMode DEFAULT_PATH_NODE_MODE=NodeMode.PERSISTENT;
	
	private final static NodeMode DEFAULT_SERVER_NODE_MODE=NodeMode.EPHEMERAL;
	
	private final static NodeMode DEFAULT_CLIENT_NODE_MODE=NodeMode.PERSISTENT;
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private RegistryConfig config;
	private String root;
	
	public AbstractRegistry() {
		
	}

	public AbstractRegistry(RegistryConfig config) {
		this.config=config;
		this.root=config.getRoot();
	}
	
	@Override
	public void init(RedisCluster redisCluster) {
		initPath();
	}
	
	@Override
	public void destroy() {
	}
	
	protected void initPath() {
		if(!isExist("")) {
			register("", new Node(null,NodeMode.PERSISTENT));
		}
		
		if(!isExist(Configs.SERVER_REGISTRY_SERVER_PATH)) {
			register(Configs.SERVER_REGISTRY_SERVER_PATH, new Node(null,DEFAULT_PATH_NODE_MODE));
		}
		
		if(!isExist(Configs.SERVER_REGISTRY_CLIENT_PATH)) {
			register(Configs.SERVER_REGISTRY_CLIENT_PATH, new Node(null,DEFAULT_PATH_NODE_MODE));
		}
	}

	@Override
	public String getServerPath() {
		return getServerPath("");
	}
	
	@Override
	public String getServerPath(ServerConfig serverConfig) {
		return getServerPath(getSerializedPath(serverConfig));
	}
	
	@Override
	public boolean registerServer(ServerConfig serverConfig) {
		return register(getServerPath(), serverConfig,new Node(null,DEFAULT_SERVER_NODE_MODE));
	}

	@Override
	public boolean unregisterServer(ServerConfig serverConfig) {
		return unregister(getServerPath(), serverConfig);
	}

	@Override
	public Collection<ServerConfig> getServers() {
		Collection<Node> nodes=getNodes(getServerPath(), ServerConfig.class);
		if(CollectionUtils.isEmpty(nodes)) {
			return Collections.emptyList();
		}
		
		Collection<ServerConfig> result=new ArrayList<ServerConfig>(nodes.size());
		for(Node node:nodes) {
			result.add(node.getData());
		}
		
		return result;
	}

	@Override
	public boolean listenServer(Listener listener) {
		return listen(getServerPath(), listener);
	}
	
	@Override
	public String getClientPath() {
		return getClientPath("");
	}
	
	@Override
	public String getClientPath(ClientConfig clientConfig) {
		return getClientPath(getSerializedPath(clientConfig));
	}

	@Override
	public boolean registerClient(ClientConfig clientConfig) {
		return register(getClientPath(), clientConfig ,new Node(clientConfig.getInfo(),DEFAULT_CLIENT_NODE_MODE));
	}

	@Override
	public boolean unregisterClient(ClientConfig clientConfig) {
		return unregister(getClientPath(), clientConfig);
	}

	@Override
	public Collection<ClientConfig> getClients() {
		Collection<Node> nodes=getNodes(getClientPath(), ClientConfig.class);
		if(CollectionUtils.isEmpty(nodes)) {
			return Collections.emptyList();
		}
		
		Collection<ClientConfig> result=new ArrayList<ClientConfig>(nodes.size());
		for(Node node:nodes) {
			ClientConfig config=node.getData();
			config.setInfo(getNode(getClientPath(), config).getData());
			result.add(config);
		}
		
		return result;
	}

	@Override
	public boolean listenClient(Listener listener) {
		return listen(getClientPath(), listener);
	}
	
	@Override
	public boolean register(String path, Object value,Node node) {
		return register(getSerializedPath(path,value), node);
	}
	
	@Override
	public boolean unregister(String path, Object value) {
		return unregister(getSerializedPath(path,value));
	}
	
	@Override
	public boolean unregister(String path) {
		if(logger.isDebugEnabled()) {
			logger.debug("取消注册 - {}",path);
		}
		
		try {
			return doUnregister(getRootPath(path));
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	@Override
	public Node getNode(String path, Class<?> type) {
		Node node=getNode(path);
		if(node==null) {
			return null;
		}

		Object data=doDeserialize(node.getPath(), type);
		node.setData(data);
		return node;
	}
	
	@Override
	public Collection<Node> getNodes(String path, Class<?> type) {
		Collection<Node> nodes=getNodes(path);
		if(CollectionUtils.isEmpty(nodes)) {
			return Collections.emptyList();
		}
		
		for(Node node:nodes) {
			Object data=doDeserialize(node.getPath(), type);
			node.setData(data);
		}
		
		return nodes;
	}
	
	@Override
	public boolean register(String path) {
		return register(path,null);
	}
	
	@Override
	public boolean register(String path, Node node) {
		if(isExist(path)) {
			return setNode(path, node);
		}
		else {
			if(logger.isDebugEnabled()) {
				logger.debug("注册 - {} - {}",path,node);
			}
			
			try {
				return doRegister(getRootPath(path),node);
			}
			catch(Exception e) {
				throw new RegistryException(e);
			}
		}
	}
	
	@Override
	public boolean setNode(String path, Object value, Node node) {
		return setNode(getSerializedPath(path, value), node);
	}
	
	@Override
	public boolean setNode(String path, Node node) {
		if(!isExist(path)) {
			return false;
		}
		
		if(logger.isDebugEnabled()) {
			logger.debug("设置节点 - {} - {}",path,node);
		}
		
		try {
			return doSetNode(getRootPath(path), node);
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	@Override
	public Node getNode(String path, Object value) {
		return getNode(getSerializedPath(path, value));
	}
	
	@Override
	public Node getNode(String path, Object value, Class<?> type) {
		return getNode(getSerializedPath(path, value), type);
	}
	
	@Override
	public Collection<Node> getNodes(String path, Object value, Class<?> type) {
		return getNodes(getSerializedPath(path, value), type);
	}
	
	@Override
	public Node getNode(String path) {
		if(!isExist(path)) {
			return null;
		}
		
		try {
			return doGetNode(getRootPath(path));
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	@Override
	public Collection<Node> getNodes(String path) {
		try {
			return doGetNodes(getRootPath(path));
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	@Override
	public boolean isExist(String path) {
		try {
			return doIsExist(getRootPath(path));
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	@Override
	public boolean listen(String path, Listener listener) {
		if(logger.isDebugEnabled()) {
			logger.debug("监听节点 - {}",path);
		}
		
		try {
			return doListen(getRootPath(path), listener);
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	@Override
	public boolean unlisten(String path, Listener listener) {
		if(logger.isDebugEnabled()) {
			logger.debug("取消监听节点 - {}",path);
		}
		
		try {
			return doUnlisten(getRootPath(path), listener);
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	@Override
	public boolean selectLeader(String path, int timeout, Listener listener) {
		if(logger.isDebugEnabled()) {
			logger.debug("选举leader - {}",path);
		}
		
		try {
			return doSelectLeader(getRootPath(path), timeout, listener);
		}
		catch(Exception e) {
			throw new RegistryException(e);
		}
	}
	
	protected String doSerialize(Object value) {
		return JSON.toJSONString(value);
	}
	
	protected <T> T doDeserialize(String value,Class<T> type) {
		return JSON.parseObject(value,type);
	}
	
	protected byte[] doByteSerialize(Object value) {
		return JSON.toJSONBytes(value,SerializerFeature.WriteClassName);
	}
	
	protected <T> T doByteDeserialize(byte value[]) {
		return (T) JSON.parse(value);
	}
	
	protected <T> T doByteDeserialize(byte value[],Class<T> type) {
		return JSON.parseObject(value,type);
	}
	
	protected String getSerializedPath(Object value) {
		return doSerialize(value);
	}
	
	protected String getSerializedPath(String path,Object value) {
		return getPath(path, doSerialize(value));
	}
	
	protected String getServerPath(String path) {
		return getPath(Configs.SERVER_REGISTRY_SERVER_PATH, path);
	}
	
	protected String getClientPath(String path) {
		return getPath(Configs.SERVER_REGISTRY_CLIENT_PATH, path);
	}
	
	protected String getPath(String path,String node) {
		return getPath(null,path, node);
	}
	
	protected String getRootPath(String path) {
		return getPath(root,path,null);
	}
	
	protected String getPath(String root,String path,String node) {
		if(StringUtils.isBlank(root)) {
			root="";
		}
		
		if(StringUtils.isBlank(path)) {
			return root;
		}

		if(StringUtils.isBlank(path)) {
			path="";
		}
		else {
			if(!path.startsWith(Configs.DEFAULT_REGISTRY_CONNECTOR)) {
				path=Configs.DEFAULT_REGISTRY_CONNECTOR+path;
			}
		}
		
		if(StringUtils.isBlank(node)) {
			node="";
		}
		else {
			if(!node.startsWith(Configs.DEFAULT_REGISTRY_CONNECTOR)) {
				node=Configs.DEFAULT_REGISTRY_CONNECTOR+node;
			}
		}
		
		return root+path+node;
	}
	
	protected NodeMode getNodeMode(Node node) {
		if(node==null) {
			return DEFAULT_NODE_MODE;
		}
		
		if(node.getMode()==null) {
			return DEFAULT_NODE_MODE;
		}
		
		return node.getMode();
	}
	
	protected abstract boolean doRegister(String path, Node node) throws Exception;
	
	protected abstract boolean doUnregister(String path) throws Exception;
	
	protected abstract boolean doSetNode(String path, Node node) throws Exception;
	
	protected abstract Node doGetNode(String path) throws Exception;
	
	protected abstract Collection<Node> doGetNodes(String path) throws Exception;
	
	protected abstract boolean doIsExist(String path) throws Exception;
	
	protected abstract boolean doListen(String path, Listener listener) throws Exception;
	
	protected abstract boolean doUnlisten(String path, Listener listener) throws Exception;
	
	protected abstract boolean doSelectLeader(String path, int timeout, Listener listener) throws Exception;
	
	public void setRoot(String root) {
		this.root = root;
	}
	public String getRoot() {
		return root;
	}
}